package promote.brozen.rsocket.channel;

import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.core.RSocketServer;
import io.rsocket.core.Resume;
import io.rsocket.transport.netty.server.CloseableChannel;
import io.rsocket.transport.netty.server.TcpServerTransport;
import io.rsocket.util.DefaultPayload;
import org.apache.commons.lang3.StringUtils;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.concurrent.locks.LockSupport;

/**
 * @author Brozen
 * @date 2020/6/28 9:30 AM
 * @email brozen@qq.com
 */
public class ChannelServer {

    public static void main(String[] args) {
        // 如果用websocket，可以用WebsocketServerTransport
        TcpServerTransport tcpTransport = TcpServerTransport.create("localhost", 8009);

        CloseableChannel channel = RSocketServer.create()
                .resume(new Resume())
                .acceptor((setup, rSocket) -> {
                    // rSocket 是客户端的RSocket
                    System.out.println("[Server] new connection created");
                    return Mono.just(new EchoServer());
                })
                .bind(tcpTransport)
                .block();
        System.out.println("server启动" + channel.address());

        LockSupport.park();
    }

    static class EchoServer implements RSocket {

        public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
            System.out.println("接收新的客户端");
            return Flux.from(payloads)
                    .switchOnFirst((signal, flux) -> {
                        if (signal.hasValue()) {
                            Payload loginPayload = signal.get();
                            String username = loginPayload.getMetadataUtf8();
                            System.out.println("用户登录");

                            if (StringUtils.isBlank(username)) {
                                return Flux.error(new IllegalArgumentException("用户名为空"));
                            } else {
                                EchoController controller = new EchoController(username);
                                flux.subscribe(controller);
                                return Flux.just(DefaultPayload.create("登录成功")).concatWith(Flux.from(controller));
                            }

                        } else if (signal.hasError()) {
                            return Flux.error(signal.getThrowable());
                        } else {
                            return Flux.empty();
                        }
                    });
        }
    }

    static class EchoController extends BaseSubscriber<Payload> implements Publisher<Payload> {

        private String username;

        private Subscriber<? super Payload> subscriber;

        public EchoController(String username) {
            this.username = username;
        }

        private void saveSend(String msg) {
            if (this.subscriber != null) {
                this.subscriber.onNext(DefaultPayload.create("{\"msg\": "+msg+"}", this.username));
            }
        }

        @Override
        public void subscribe(Subscriber<? super Payload> s) {
            this.subscriber = s;
        }

        @Override
        protected void hookOnNext(Payload payload) {
            String data = payload.getDataUtf8();
            System.out.println("[server receive] " + data);
            saveSend("i got you : " + data);
        }

        @Override
        protected void hookOnError(Throwable t) {
            t.printStackTrace();
        }

        @Override
        protected void hookOnComplete() {
            System.out.println("[server] complete");
        }
    }

}
